<?php
/**
 * Created by PhpStorm.
 * User: bzg
 * Date: 2019/12/3
 * Time: 15:30
 */

namespace App\Extend\SysExtend\Queue\Driver;


use App\Extend\SysExtend\Helper;
use EasySwoole\Queue\Job;
use EasySwoole\Queue\QueueDriverInterface;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

class RabbitMQ implements QueueDriverInterface
{
    /**
     * 连接
     * @var AMQPStreamConnection
     */
    protected $con;

    /**
     * 消息
     * @var array
     */
    protected $message = [];

    /**
     * 通道
     * @var \PhpAmqpLib\Channel\AMQPChannel
     */
    protected $channel;

    /**
     * @var string 队列名称
     */
    protected $queueName;

    protected $config = [
        'host' => '192.168.3.139',
        'port' => 5672,
        'user' => 'guest',
        'password' => 'guest',
        'default_name' => 'rabbit'
    ];

    public function __construct(string $queueName = '', $conf = [])
    {
        if($conf) {
            $this->config = $conf;
        }
        $this->queueName = $queueName ? $queueName : $this->config['default_name'];
        // 连接rabbitMq
        $this->con = new AMQPStreamConnection(
            $this->config['host'],
            $this->config['port'],
            $this->config['user'],
            $this->config['password']
        );
        //}
        // 获取通道
        $this->channel = $this->con->channel();
        // 声明队列
        $this->channel->queue_declare($this->queueName,false,false,false,false);
    }

    public function push(Job $job): bool
    {

        $data = $job->__toString();
        // 组装消息
        $msg = new AMQPMessage($data);
        $this->channel->basic_publish($msg,'',$this->queueName);
        return true;
    }

    public function pop(float $timeout = 3.0): ?Job
    {
        // 消费者
        $callback = function (AMQPMessage $msg){
            $data = json_decode($msg->getBody(),true);
            if(is_array($data)){
                array_unshift($this->message,new Job($data)) ;
            }else{
                return null;
            }
        };

        $this->channel
            ->basic_consume($this->queueName, '', false, true, false, false, $callback);

        while ($this->channel->is_consuming()) {
            if($this->message) {
                return array_pop($this->message);
            }
            $this->channel->wait();
        }
    }

    public function size(): ?int
    {
        return 1;
    }
}